package flink;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import static org.apache.kafka.clients.consumer.ConsumerConfig.*;

@Slf4j
public class KafkaConsumerApp {
    public static void main(String[] args) {
//        String name = System.getProperty("name");
//        System.out.println("name");
//        System.out.println(name);
        try {

            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            Properties properties = new Properties();
            properties.setProperty(BOOTSTRAP_SERVERS_CONFIG, "101.37.152.195:9092");
            properties.setProperty(GROUP_ID_CONFIG, "zhTestGroup");


            FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer("userTopic", new SimpleStringSchema(), properties);
            // 设置消费位置-从最新位置
            // flinkKafkaConsumer.setStartFromLatest();
            flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(true);


            DataStream<String> stream = env
                    .addSource(flinkKafkaConsumer)
                    .setParallelism(1);

            // 水印策略
            WatermarkStrategy<String> strategy = WatermarkStrategy
                    .<String>forMonotonousTimestamps();

            stream
                    .assignTimestampsAndWatermarks(strategy)
                    // 翻转事件时间窗口 5秒内的消息
                    .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                    .apply(new MyAllWindowFunction())
                    .addSink(new MySinkFunction());

            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class MySinkFunction extends RichSinkFunction<List<String>> {

    @Override
    public void invoke(List<String> value, Context context) {
        //遍历数据集合
        for (String student : value) {
            System.out.println("数据：" + student);
        }
    }
}

class MyAllWindowFunction implements AllWindowFunction<String, List<String>, TimeWindow> {

    @Override
    public void apply(TimeWindow window, Iterable<String> values, Collector<List<String>> out) {
        ArrayList<String> students = Lists.newArrayList(values);
        System.out.println("时间" + System.currentTimeMillis() / 1000 + "-5秒内收集到的数据条数是：" + students.size());
        out.collect(students);
    }
}